package com.aotain.project.apollo.bolt;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;
import com.aotain.storm.TupleHelpers;

/**
 * 
 * @ClassName: AbnStatisBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2016年4月28日 下午1:33:32 
 *
 */
public class AbnSourceStatisBolt extends AbstractBolt{

	private static final long serialVersionUID = 1L;

	String dip = null;
	String sip = null;
	String dateStr_d = null;
	String areaName = null;
	String abnAreaMapKey = null;
	Long triplePV = null;
	String destcity = null;
	String proPVKey = null;
	String dprovinceId = null;
	Map<String,Long> abnAreaMap = new HashMap<String,Long>();
	Map<String,Long> proPVMap = new HashMap<String,Long>();




	@Override
	public void execute(Tuple tuple) {
		try {
			if(TupleHelpers.isTickTuple(tuple)) {
				emitCountingData(collector);  	
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			System.err.println(" AbnStatisBolt exception"+e.getMessage());
			Logger.getRootLogger().error("AbnStatisBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException(e);
		}	
	}

	private void emitCountingData(OutputCollector collector) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);   

		String rowkeyAbnArea = null;
		String proAbnKey = null;

		for(Entry<String,Long> entry : abnAreaMap.entrySet()) {
			String key = entry.getKey();
			long abnAreaPV = entry.getValue().longValue();
			String[] splits1 = key.split("_",-1);
			proAbnKey = splits1[0] + "_" + splits1[1] + "_" +splits1[5]; 
			rowkeyAbnArea = splits1[0] + "_" + splits1[1];//sip_date
			
			//			System.err.println("SDS_ABNORMAL_AREA......PV=====" + abnAreaPV + "key is " + key + " IP is " + splits[0] + " REPORTTIME is " + splits[1] + " AREA is " + splits[3]);
			hbaseInstance.Add("SDS_ABN_REQ_AREA", rowkeyAbnArea, "cf", "IP", splits1[0]);
			hbaseInstance.Add("SDS_ABN_REQ_AREA", rowkeyAbnArea, "cf", "REPORTTIME", splits1[1]);
			hbaseInstance.Add("SDS_ABN_REQ_AREA", rowkeyAbnArea, "cf", "AREA", splits1[3]);
			hbaseInstance.Add("SDS_ABN_REQ_AREA", rowkeyAbnArea, "cf", "CITYNAME", splits1[4]);
			hbaseInstance.incerment("SDS_ABN_REQ_AREA", rowkeyAbnArea, "cf:PV", abnAreaPV);
			
			//检测IP发送异常连接到省的统计 
			int stpv = proPVMap.get(proAbnKey).intValue();
			hbaseInstance.incerment("SDS_ABNORMAL_AREA", proAbnKey, "cf:STPROVINCEPV", stpv);



		}

		proPVMap.clear();
		abnAreaMap.clear();
	}

	private void countInLocal(Tuple tuple) {

		if(tuple != null ) {  
			dip = tuple.getStringByField("dip");
			sip = tuple.getStringByField("sip");
			dateStr_d = tuple.getStringByField("dateStr_d"); 
			areaName = tuple.getStringByField("destarea");
			destcity = tuple.getStringByField("destcity");
			triplePV = tuple.getLongByField("pv");
			dprovinceId = tuple.getStringByField("dprovinceId");
			abnAreaMapKey = sip + "_" + dateStr_d + "_" + dip +  "_" + areaName + "_" + destcity + "_" + dprovinceId;
			proPVKey = sip + "_" + dateStr_d +"_" + dprovinceId;

			long abnAreaPV = triplePV;
			long abnProPV = triplePV;

			if(abnAreaMap.get(abnAreaMapKey) != null) {
				abnAreaPV = abnAreaMap.get(abnAreaMapKey) + triplePV;
			}
			if(proPVMap.get(proPVKey) != null) {
				abnProPV = proPVMap.get(proPVKey) + triplePV;
			}
			abnAreaMap.put(abnAreaMapKey, abnAreaPV);
			proPVMap.put(proPVKey, abnProPV);

		} 
	}

	@Override
	public void cleanup() {
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,900);
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
	}

}
